package com.mercury.kafka.listener;

import com.mercury.kafka.base.BaseListener;
import com.mercury.kafka.domain.KafKaMqMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @author mercury
 */
@Component
@Slf4j
public class KafkaMercuryListener extends BaseListener {


    @Override
    protected void execute(ConsumerRecord<String, KafKaMqMessage> content) {
        log.info("execute success");
    }

    @Override
    protected void failExecute(ConsumerRecord<String, KafKaMqMessage> content) {
        log.info("execute failExecute");
    }

    @KafkaListener(
            groupId = "${mercury.kafka.groupId:topic-mercury}",
            topics = "${mercury.kafka.topic:topic-groupId}")
    public void process(ConsumerRecord<String, KafKaMqMessage> record, Acknowledgment acknowledgment) {
        log.info("kafka接收的生产者消息为: topic = {" + record.topic()
                + "}, offset = {" + record.offset() + "}, value = {" + record.value() + "}");
        super.receiveMessage(record, acknowledgment);
    }

}
